/**
 * Copyright (C) 2008 by The Regents of the University of California
 * Redistribution of this file is permitted under the terms of the GNU
 * Public License (GPL).
 *
 * @author Junghoo "John" Cho <cho AT cs.ucla.edu>
 * @date 3/24/2008
 */

#include <cstdio>
#include <iostream>
#include <fstream>
#include "Bruinbase.h"
#include "SqlEngine.h"
#include "BTreeIndex.h"

using namespace std;

// external functions and variables for load file and sql command parsing 
extern FILE* sqlin;
int sqlparse(void);


RC SqlEngine::run(FILE* commandline)
{
  fprintf(stdout, "Bruinbase> ");

  // set the command line input and start parsing user input
  sqlin = commandline;
  sqlparse();  // sqlparse() is defined in SqlParser.tab.c generated from
               // SqlParser.y by bison (bison is GNU equivalent of yacc)

  return 0;
}

/*
    attr=1 means "key" 
    attr=2 means "value" 
    attr=3 means "*" 
    attr=4 means "COUNT(*)
*/

RC SqlEngine::select(int attr, const string& table, const vector<SelCond>& cond)
{
    RecordFile rf;   // RecordFile containing the table
    RecordId   rid;  // record cursor for table scanning

    RC     rc;
    int    key;     
    string value;
    int    count;
    int    diff;

    // open the table file
    if ((rc = rf.open(table + ".tbl", 'r')) < 0) {
        fprintf(stderr, "Error: table %s does not exist\n", table.c_str());
        return rc;
    }

    IndexCursor cursor;
    BTreeIndex index;

    vector<int> not_equals;    // a key that is not equal
    vector<SelCond> str_vals;         // cond.val = 2, keys that are equal
    set<int> val_keys;
    string v;
    int k;

    int upper_bound = INT_MAX;
    int lower_bound = -1;

    if (index.open(table + ".idx", 'r') == 0) {
        for (unsigned j = 0; j < cond.size(); j++) {
            if ( cond[j].attr == 1 ) {
                int curr_val = atoi(cond[j].value);

                switch (cond[j].comp) {
                case SelCond::EQ:
                    if ( curr_val == atoi(cond[j].value) ){
                        if (curr_val < upper_bound ) { upper_bound = curr_val; }
                        if (curr_val > lower_bound ) { lower_bound = curr_val; }
                    }
                    break;
                case SelCond::NE:
                    not_equals.push_back(curr_val);
                    break;
                case SelCond::GT:   // >, to the right of this number
                    if (curr_val > lower_bound) { lower_bound = curr_val+1; }
                    break;
                case SelCond::LT:   // <, to the left of this number
                    if (curr_val < upper_bound) { upper_bound = curr_val; }
                    break;
                case SelCond::GE:   // >=
                    if (curr_val > lower_bound) { lower_bound = curr_val; } 
                    break;
                case SelCond::LE:   // <=
                    if (curr_val < upper_bound) { upper_bound = curr_val+1; }
                    break;
                }   // end switch
            } else {    // cond[j].attr = 2, value, sequential search
                str_vals.push_back(cond[j]);
            }
        }

        // vars used here: v = value, k = key
        if (!str_vals.empty() && lower_bound <= upper_bound) {
            rid.pid = rid.sid = 0;
            while(rid < rf.endRid()) {
                if ((rc = rf.read(rid, k, v)) < 0) {
                    fprintf(stderr, "Error: while reading a tuple from table %s\n", table.c_str());
                    index.close();
                    rf.close();
                    return 0;
                }

                vector<SelCond>::iterator val;
                for ( val=str_vals.begin() ; val < str_vals.end(); val++ ) {
                    diff = strcmp(v.c_str(), val->value);
                    switch (val->comp) {
                        case SelCond::EQ:
                            if (diff != 0) goto nt;
                            break;
                        case SelCond::NE:
                            if (diff == 0) goto nt;
                            break;
                        case SelCond::GT:
                            if (diff <= 0) goto nt;
                            break;
                        case SelCond::LT:
                            if (diff >= 0) goto nt;
                            break;
                        case SelCond::GE:
                            if (diff < 0) goto nt;
                            break;
                        case SelCond::LE:
                            if (diff > 0) goto nt;
                        break;
                    }
                }   
                if ( k > lower_bound && k < upper_bound) {
                    val_keys.insert(k);
                }

                nt:
                ++rid;
            }
        }

        count = 0;
        //cout << "upper key " << upper_bound << " lower key " << lower_bound << endl;
        if ( lower_bound > upper_bound) { goto index_exit; }
        index.empty_stack();
        index.locate(lower_bound, cursor);
        if ( cursor.eid < 0 ) {goto index_exit;}
        
        while(1) {
            rc = index.readForward(cursor, key, rid);
            if (upper_bound == lower_bound && key == upper_bound) { rc = -1; }
            else if (key >= upper_bound) { break; }
            
            vector<int>::iterator v_it;
            bool print_line = true;
            for ( v_it=not_equals.begin() ; v_it < not_equals.end(); v_it++ ) {
                if (key == *v_it) { 
                    print_line = false; 
                    break;
                }
            }
            if (print_line && !str_vals.empty()) {
                if ( val_keys.find(key) == val_keys.end() ) { print_line = false; }
            }

            if (print_line) {
                count ++;
                switch (attr) {
                    case 1:  // SELECT key
                      fprintf(stdout, "%d\n", key);
                      break;
                    case 2:  // SELECT value
                        rf.read(rid, key, value);
                        fprintf(stdout, "%s\n", value.c_str());
                      break;
                    case 3:  // SELECT *
                        rf.read(rid, key, value);
                        fprintf(stdout, "%d '%s'\n", key, value.c_str());
                      break;
                }
            }

            if (rc < 0) {break;}
        }


        if (attr == 4) {
            fprintf(stdout, "%d\n", count);
        }

        index_exit:
        if (!str_vals.empty()) { rf.close(); }
        index.close();
        return 0;
    } else {    // no index, sequential search
        // scan the table file from the beginning
        rid.pid = rid.sid = 0;
        count = 0;
        while (rid < rf.endRid()) {
        // read the tuple
        if ((rc = rf.read(rid, key, value)) < 0) {
          fprintf(stderr, "Error: while reading a tuple from table %s\n", table.c_str());
          goto exit_select;
        }

        // check the conditions on the tuple
        for (unsigned i = 0; i < cond.size(); i++) {
          // compute the difference between the tuple value and the condition value
          switch (cond[i].attr) {
          case 1:
        diff = key - atoi(cond[i].value);
        break;
          case 2:
        diff = strcmp(value.c_str(), cond[i].value);
        break;
          }

          // skip the tuple if any condition is not met
          switch (cond[i].comp) {
          case SelCond::EQ:
        if (diff != 0) goto next_tuple;
        break;
          case SelCond::NE:
        if (diff == 0) goto next_tuple;
        break;
          case SelCond::GT:
        if (diff <= 0) goto next_tuple;
        break;
          case SelCond::LT:
        if (diff >= 0) goto next_tuple;
        break;
          case SelCond::GE:
        if (diff < 0) goto next_tuple;
        break;
          case SelCond::LE:
        if (diff > 0) goto next_tuple;
        break;
          }
        }

        // the condition is met for the tuple. 
        // increase matching tuple counter
        count++;

        // print the tuple 
        switch (attr) {
        case 1:  // SELECT key
          fprintf(stdout, "%d\n", key);
          break;
        case 2:  // SELECT value
          fprintf(stdout, "%s\n", value.c_str());
          break;
        case 3:  // SELECT *
          fprintf(stdout, "%d '%s'\n", key, value.c_str());
          break;
        }

        // move to the next tuple
        next_tuple:
        ++rid;
        }

        // print matching tuple count if "select count(*)"
        if (attr == 4) {
        fprintf(stdout, "%d\n", count);
        }
        rc = 0;

        // close the table file and return
        exit_select:
        rf.close();
        return rc;
    }
}

RC SqlEngine::load(const string& table, const string& loadfile, bool index)
{
    // new file to write to
    RecordFile new_file;
    string filename = table + ".tbl";
    new_file.open(filename, 'w');

    BTreeIndex tree_index;

    if (index) {
        tree_index.open(table + ".idx", 'w');
    }

    // file to read from
    ifstream ifs(loadfile.c_str());
    if(!ifs) {
        return RC_FILE_OPEN_FAILED;
    }

    // read file
    string buffer;
    int key;
    while ( getline(ifs, buffer, '\n') ) {
        string value; 
        RecordId rid; 
        RC rc;
        if ( (rc = parseLoadLine(buffer, key, value)) < 0 ) {return rc;}
        if ( (rc = new_file.append(key, value, rid)) < 0) {return rc;}
        if ( index ) {
            if ( (rc = tree_index.insert(key, rid) < 0) ){ 
                return rc; 
            }
        }
    }
    new_file.close();
    tree_index.close();
    return 0;
}

RC SqlEngine::parseLoadLine(const string& line, int& key, string& value)
{
    const char *s;
    char        c;
    string::size_type loc;
    
    // ignore beginning white spaces
    c = *(s = line.c_str());
    while (c == ' ' || c == '\t') { c = *++s; }

    // get the integer key value
    key = atoi(s);

    // look for comma
    s = strchr(s, ',');
    if (s == NULL) { return RC_INVALID_FILE_FORMAT; }

    // ignore white spaces
    do { c = *++s; } while (c == ' ' || c == '\t');
    
    // if there is nothing left, set the value to empty string
    if (c == 0) { 
        value.erase();
        return 0;
    }

    // is the value field delimited by ' or "?
    if (c == '\'' || c == '"') {
        s++;
    } else {
        c = '\n';
    }

    // get the value string
    value.assign(s);
    loc = value.find(c, 0);
    if (loc != string::npos) { value.erase(loc); }

    return 0;
}
